/*-------------------------------------------------------------------------
 *
 * bloomfilter.c
 *		空间高效的集合成员测试
 *
 * Bloom 过滤器是一种概率数据结构，用于测试元素的集合成员资格。可能会出现假阳性，但不会出现假阴性；对集合的成员资格测试返回“可能在集合中”或“绝对不在集合中”。这通常非常节省空间，这可能是一个决定性的优势。
 *
 * 元素可以添加到集合中，但不能删除。添加的元素越多，假阳性的概率就越大。调用者必须在初始化 Bloom 过滤器时提供集合的估计总大小。这用于在最终的假阳性率与内存的使用之间进行平衡。
 *
 * 该实现非常适合处理无序集合之间的数据同步问题，特别是在可预测的性能很重要和某些假阳性可接受的情况下。它也非常适合处理缓存过滤问题，在这些问题中，相对较小和/或低基数的集合被指纹化，特别是在许多后续的成员资格测试最终表明感兴趣的值不存在时。这应该可以节省调用者许多权威查找，例如对更大磁盘结构的昂贵探测。
 *
 * Copyright (c) 2018-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/lib/bloomfilter.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <math.h>

#include "common/hashfn.h"
#include "lib/bloomfilter.h"
#include "port/pg_bitutils.h"

#define MAX_HASH_FUNCS		10

struct bloom_filter
{
	/* K hash functions are used, seeded by caller's seed */
	int			k_hash_funcs;
	uint64		seed;
	/* m 是比特集大小，以比特为单位。必须是小于或等于 2^32 的 2 的幂。 */
	uint64		m;
	unsigned char bitset[FLEXIBLE_ARRAY_MEMBER];
};

static int	fc_my_bloom_power(uint64 fc_target_bitset_bits);
static int	fc_optimal_k(uint64 fc_bitset_bits, int64 fc_total_elems);
static void fc_k_hashes(bloom_filter *fc_filter, uint32 *fc_hashes, unsigned char *fc_elem,
					 size_t fc_len);
static inline uint32 fc_mod_m(uint32 fc_a, uint64 fc_m);

/*
 * 在调用者的内存上下文中创建布隆过滤器。当比特集大小不受内存
 * 可用性的限制时，我们目标是 1% 到 2% 的假阳性率。
 *
 * total_elems 是集合最终大小的估计。它应该
 * 大致正确，但实现可以很好地处理可能偏差五倍或更多的情况。
 * 有关原因，请参见“概率验证中的布隆过滤器”（Dillinger & Manolios，2004）。
 *
 * bloom_work_mem 的大小以 KB 为单位，符合一般的 work_mem 规范。
 * 这决定了基础比特集的大小（不计算微不足道的管理空间）。
 * 比特集的大小始终是比特的 2 的幂的数量，最大可能的比特集为 512MB
 * （2^32 比特）。该实现仅分配足够的内存以达到其标准假阳性
 * 率，使用一个简单的公式，以调用者的 total_elems 估计作为
 * 输入。即使在 bloom_work_mem 更高的情况下，比特集的大小
 * 可能也小至 1MB。
 *
 * 布隆过滤器使用调用者提供的值进行种子设置。在每次调用中
 * 使用不同的种子值，使得在第二次指纹识别相同集合时，相同的
 * 假阳性再次出现的可能性很小。不关心此问题的调用者
 * 传递一个常量作为他们的种子，通常为 0。调用者还可以使用伪随机
 * 种子，例如来自 pg_prng_uint64()。
 */
bloom_filter *
bloom_create(int64 fc_total_elems, int fc_bloom_work_mem, uint64 fc_seed)
{
	bloom_filter *fc_filter;
	int			fc_bloom_power;
	uint64		fc_bitset_bytes;
	uint64		fc_bitset_bits;

	/*
	 * 每个元素目标为两个字节；这足以使假阳性
	 * 率低于 1%，而与比特集的大小或元素总数无关。
	 * 此外，如果将比特集的大小向下舍入到下一个最低
	 * 的 2 的幂显著下降，假阳性率在几乎所有情况下
	 * 仍不会超过 2%。
	 */
	fc_bitset_bytes = Min(fc_bloom_work_mem * UINT64CONST(1024), fc_total_elems * 2);
	fc_bitset_bytes = Max(1024 * 1024, fc_bitset_bytes);

	/*
	 * 位数大小应为 <= 目标的最高二次幂。bitset_bits
	 * 是 uint64，因为 PG_UINT32_MAX 为 2^32 - 1，而不是 2^32。
	 */
	fc_bloom_power = fc_my_bloom_power(fc_bitset_bytes * BITS_PER_BYTE);
	fc_bitset_bits = UINT64CONST(1) << fc_bloom_power;
	fc_bitset_bytes = fc_bitset_bits / BITS_PER_BYTE;

	/* 分配未设置比特集的布隆过滤器 */
	fc_filter = palloc0(offsetof(bloom_filter, bitset) +
					 sizeof(unsigned char) * fc_bitset_bytes);
	fc_filter->k_hash_funcs = fc_optimal_k(fc_bitset_bits, fc_total_elems);
	fc_filter->seed = fc_seed;
	fc_filter->m = fc_bitset_bits;

	return fc_filter;
}

/*
 * 释放布隆过滤器
 */
void bloom_free(bloom_filter *fc_filter)
{
	pfree(fc_filter);
}

/*
 * 将元素添加到布隆过滤器
 */
void bloom_add_element(bloom_filter *fc_filter, unsigned char *fc_elem, size_t fc_len)
{
	uint32		fc_hashes[MAX_HASH_FUNCS];
	int			fc_i;

	fc_k_hashes(fc_filter, fc_hashes, fc_elem, fc_len);

	/* 将按位地址映射到字节地址 + 位偏移量 */
	for (fc_i = 0; fc_i < fc_filter->k_hash_funcs; fc_i++)
	{
		fc_filter->bitset[fc_hashes[fc_i] >> 3] |= 1 << (fc_hashes[fc_i] & 7);
	}
}

/*
 * 测试布隆过滤器是否确实缺少元素。
 *
 * 如果元素确实不在 bloom_add_element() 观察到的元素集合中，则返回 true。
 * 否则返回 false，表示元素可能在集合中存在。
 */
bool bloom_lacks_element(bloom_filter *fc_filter, unsigned char *fc_elem, size_t fc_len)
{
	uint32		fc_hashes[MAX_HASH_FUNCS];
	int			fc_i;

	fc_k_hashes(fc_filter, fc_hashes, fc_elem, fc_len);

	/* 将按位地址映射到字节地址 + 位偏移量 */
	for (fc_i = 0; fc_i < fc_filter->k_hash_funcs; fc_i++)
	{
		if (!(fc_filter->bitset[fc_hashes[fc_i] >> 3] & (1 << (fc_hashes[fc_i] & 7))))
			return true;
	}

	return false;
}

/*
 * 当前设置的比特比例是多少？
 *
 * 返回比例，以过滤器大小的乘数表示。通常应该接近0.5，即使在我们有足够的内存来保证假阳性率在目标1%到2%的范围内时也是如此，因为随着每个元素可用内存的增加，将使用更多的哈希函数。
 *
 * 这是唯一的低开销检测方法，可以出现在调试跟踪中。在调试布隆过滤器代码时，直接测试假阳性率很可能要有趣得多。
 */
double bloom_prop_bits_set(bloom_filter *fc_filter)
{
	int			fc_bitset_bytes = fc_filter->m / BITS_PER_BYTE;
	uint64		fc_bits_set = pg_popcount((char *) fc_filter->bitset, fc_bitset_bytes);

	return fc_bits_set / (double) fc_filter->m;
}

/*
 * 在二的幂序列中，哪个元素小于或等于
 * target_bitset_bits？
 *
 * 此处返回的值通常必须是安全的，作为实际位集大小的基础。
 *
 * 位集永远不能超过2 ^ 32位（512MB）。这对于所有当前调用者的需求来说是足够的，并且允许我们使用32位哈希函数。这也使我们能够轻松保持在MaxAllocSize限制之下（调用者需要为出现在灵活数组成员之前的非位集字段留出空间，因此一个1GB的位集将使用一个刚好超过MaxAllocSize的分配）。
 */
static int fc_my_bloom_power(uint64 fc_target_bitset_bits)
{
	int			fc_bloom_power = -1;

	while (fc_target_bitset_bits > 0 && fc_bloom_power < 32)
	{
		fc_bloom_power++;
		fc_target_bitset_bits >>= 1;
	}

	return fc_bloom_power;
}

/*
 * 根据位过滤器的大小（以位为单位）和预计的元素总数，确定最佳哈希函数数量。最佳数量是最小化假阳性率的数量。
 */
static int fc_optimal_k(uint64 fc_bitset_bits, int64 fc_total_elems)
{
	int			fc_k = rint(log(2.0) * fc_bitset_bits / fc_total_elems);

	return Max(1, Min(fc_k, MAX_HASH_FUNCS));
}

/*
 * 为元素生成k个哈希值。
 *
 * 调用者传递一个数组，该数组被填充由哈希调用者的元素确定的k个值。
 *
 * 实际上只使用了2个独立的真实哈希函数来支持最多MAX_HASH_FUNCS个哈希函数的接口；使用增强的双重哈希来实现这一点。我们偏向于采用增强的双重哈希而不是经典的双重哈希，其主要原因是后者在使用二的幂大小的位集时存在冲突问题。有关详细信息，请参见Dillinger & Manolios。
 */
static void fc_k_hashes(bloom_filter *fc_filter, uint32 *fc_hashes, unsigned char *fc_elem, size_t fc_len)
{
	uint64		fc_hash;
	uint32		fc_x,
				fc_y;
	uint64		fc_m;
	int			fc_i;

	/* 使用64位哈希获得两个独立的32位哈希 */
	fc_hash = DatumGetUInt64(hash_any_extended(fc_elem, fc_len, fc_filter->seed));
	fc_x = (uint32) fc_hash;
	fc_y = (uint32) (fc_hash >> 32);
	fc_m = fc_filter->m;

	fc_x = fc_mod_m(fc_x, fc_m);
	fc_y = fc_mod_m(fc_y, fc_m);

	/* 累加哈希 */
	fc_hashes[0] = fc_x;
	for (fc_i = 1; fc_i < fc_filter->k_hash_funcs; fc_i++)
	{
		fc_x = fc_mod_m(fc_x + fc_y, fc_m);
		fc_y = fc_mod_m(fc_y + fc_i, fc_m);

		fc_hashes[fc_i] = fc_x;
	}
}

/*
 * 以低成本计算“val MOD m”。
 *
 * 假定m（即位集大小）是二的幂。
 *
 * 使用二的幂的位数作为位集大小允许我们使用按位与运算来计算哈希值的模。它也是避免模偏差效应的一个简单方法。
 */
static inline uint32 fc_mod_m(uint32 fc_val, uint64 fc_m)
{
	Assert(fc_m <= PG_UINT32_MAX + UINT64CONST(1));
	Assert(((fc_m - 1) & fc_m) == 0);

	return fc_val & (fc_m - 1);
}
